package org.snail.proxy.client;

import com.google.common.collect.Maps;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.snail.proxy.client.config.AgentConfig;
import org.snail.proxy.proto.*;

import java.util.Map;

public class ProxyClient implements Client {
    private static final Logger log = LoggerFactory.getLogger(ProxyClient.class);

    private final Map<String, Client> delegates = Maps.newHashMap();
    private final AgentConfig agentConfig;

    private EventLoopGroup group;
    private Channel channel;

    public ProxyClient(AgentConfig agentConfig) {
        this.agentConfig = agentConfig;
    }

    @Override
    public void connect(String ip, int port) {
        try {
            if (group != null) {
                group.shutdownGracefully();
                group = null;
            }
            if (channel != null) {
                channel.closeFuture();
                channel = null;
            }

            group = new NioEventLoopGroup();
            //发起异步连接请求，绑定连接端口和host信息，使用NioSocketChannel来作为连接用的channel类
            ChannelFuture future = new Bootstrap().group(group).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() { // 绑定连接初始化器
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new PackageSpliter())
                                    .addLast(new ProtoDecoder(8192))
                                    .addLast(new ProtoEncoder(8192))
                                    .addLast(new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelInactive(ChannelHandlerContext ctx) {
                                            log.info("channelInactive.");
                                            System.exit(0);
                                        }

                                        @Override
                                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                            handleMessage((Packet) msg);
                                            super.channelRead(ctx, msg);
                                        }
                                    });
                        }
                    })
                    .connect(ip, port)
                    .sync();

            future.addListener((ChannelFutureListener) arg0 -> {
                if (future.isSuccess()) {
                    log.info("Connect proxy server success: {}:{}.", ip, port);
                } else {
                    log.error("Connect proxy server failed " + ip + ": " + port + ": ", future.cause());
                    group.shutdownGracefully(); //关闭线程组
                }
            });

            this.channel = future.channel();
        } catch (Exception e) {
            log.error("Connect proxy server failed: ", e);
        }
    }

    private void handleMessage(Packet packet) throws Exception {
        Client delegate;
        switch (packet.getCmd()) {
            case MessageId.CREATE_CLIENT_REQUEST_VALUE:
                CreateClientRequest createClientRequest = CreateClientRequest.parseFrom(packet.getBytes());
                delegate = NetType.TCP.name().equals(agentConfig.getType()) ?
                        new LocalTCPClient(channel, createClientRequest.getId()) : new LocalHTTPClient(channel, createClientRequest.getId(), agentConfig);
                delegate.connect(agentConfig.getLocalIp(), agentConfig.getLocalPort());
                delegates.put(createClientRequest.getId(), delegate);
                log.info("Add delegate: {}[{}] count={}.", createClientRequest.getId(), delegate.getClass().getSimpleName(), delegates.size());
                break;
            case MessageId.PROXY_MESSAGE_VALUE:
                ProxyMessage proxyMessage = ProxyMessage.parseFrom(packet.getBytes());
                delegate = delegates.get(proxyMessage.getId());
                if (delegate == null) {
                    log.error("Receive null delegate proxy server message: {}[{}].", proxyMessage.getId(), packet.getCmd());
                } else {
                    delegate.sendMessage(proxyMessage.getData().toByteArray());
                }
                break;
            case MessageId.DELETE_CLIENT_REQUEST_VALUE:
                DeleteClientRequest deleteClientRequest = DeleteClientRequest.parseFrom(packet.getBytes());
                delegate = delegates.remove(deleteClientRequest.getId());
                if (delegate != null) {
                    delegate.disconnect();
                    log.info("Remove delegate: {}[{}] count={}.", deleteClientRequest.getId(), delegate.getClass().getSimpleName(), delegates.size());
                } else {
                    log.info("Remove non-exists delegate: {} count={}.", deleteClientRequest.getId(), delegates.size());
                }
                break;
        }
        log.info("Receive proxy server message: {}[{}].", packet.getCmd(), packet.getBytes().length);
    }

    @Override
    public void disconnect() {
        channel.close();
    }

    @Override
    public void sendMessage(Object message) {
        if (channel == null || message == null || !channel.isWritable()) {
            return;
        }
        channel.writeAndFlush(message);
        log.info("ProxyClient sendMessage: {}.", message.getClass().getSimpleName());
    }

    @Override
    public boolean isConnected() {
        return channel != null && channel.isActive();
    }
}
